package com.learn.state;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.concurrent.TimeUnit;

/**
 * @create: 2023-04-22 11:00
 * @author: Mr.Du
 * --------------
 * @notes: 使用OperatorState演示类似于kafka消费数据的功能
 **/
public class OperatorStateDemo {
    public static void main(String[] args) {

    }
}

class MySourceWithState implements SourceFunction<Integer>, CheckpointedFunction {
    //定义数据是否循环生成
    private Boolean isRunning = true;
    private Integer currentCounter = 0;
    //定义ListState保存结果数据，保存offset的累加值
    private ListState<Integer> listState = null;

    /**
     * 将state中的数据持久化存储到文件（5s进行一次快照，将state数据存储到hdfs）
     * @param context
     * @throws Exception
     */
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        System.out.println("调用snapshotState方法。。。");
        //清除历史状态中存储的历史数据
        this.listState.clear();

        //将最新的累加值添加到状态中
        this.listState.add(this.currentCounter);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        System.out.println("调用initializeState方法。。。。。。。。");
        //初始化一个listState
        OperatorStateStore stateStore = context.getOperatorStateStore();
        //初始化listState
        listState = stateStore.getListState(
                new ListStateDescriptor("operator-states", TypeInformation.of(new TypeHint<Integer>() {
                })));

        //之前可能已经存储了历史数据，所以要讲历史数据获取出来继续累加
        for(Integer counter : this.listState.get()){
            //将历史存储的累加值取出来赋值给当前累加值变量
            this.currentCounter = counter;
        }

        //清除状态中存储的历史数据
        this.listState.clear();
    }

    @Override
    public void run(SourceContext<Integer> ctx) throws Exception {
            while (isRunning){
                currentCounter ++;
                ctx.collect(currentCounter);

                TimeUnit.SECONDS.sleep(1L);
                if(this.currentCounter == 10){
                    System.out.println("手动抛出异常"+(1/0));
                }
            }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
